package com.brainwave.data;

import com.brainwave.models.BrainWaveData;
import com.brainwave.bluetooth.NeuroSkyProtocol;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.subjects.PublishSubject;

/**
 * 数据处理器类
 * 负责实时处理从NeuroSky设备接收到的脑波数据
 * 包括数据缓冲、滤波、分析和状态监控
 */
public class DataProcessor implements NeuroSkyProtocol.NeuroSkyDataListener {
    
    private static final String TAG = "DataProcessor";
    
    // 数据缓冲区大小
    private static final int BUFFER_SIZE = 1000;
    private static final int RAW_BUFFER_SIZE = 512;
    
    // 数据质量阈值
    private static final int POOR_SIGNAL_THRESHOLD = 50;
    private static final int MIN_VALID_SAMPLES = 10;
    
    // 数据缓冲队列
    private final ConcurrentLinkedQueue<BrainWaveData> dataBuffer;
    private final ConcurrentLinkedQueue<Integer> rawDataBuffer;
    
    // 数据统计
    private volatile int totalSamples = 0;
    private volatile int validSamples = 0;
    private volatile long lastDataTime = 0;
    private volatile boolean isProcessing = false;
    
    // 当前会话信息
    private String currentSessionId;
    private String deviceName;
    private String deviceAddress;
    
    // 线程池
    private final ExecutorService executorService;
    
    // RxJava Subject用于数据流
    private final PublishSubject<BrainWaveData> dataSubject;
    private final PublishSubject<ProcessedData> processedDataSubject;
    private final PublishSubject<DataQualityInfo> qualitySubject;
    
    // 数据监听器接口
    public interface DataProcessorListener {
        void onDataProcessed(BrainWaveData data);
        void onProcessedDataReady(ProcessedData processedData);
        void onDataQualityChanged(DataQualityInfo qualityInfo);
        void onProcessingError(String error);
    }
    
    private DataProcessorListener listener;
    
    /**
     * 处理后的数据类
     */
    public static class ProcessedData {
        public final long timestamp;
        public final double averageAttention;
        public final double averageMeditation;
        public final double relaxationIndex;
        public final double focusIndex;
        public final double signalStability;
        public final int sampleCount;
        
        public ProcessedData(long timestamp, double averageAttention, double averageMeditation,
                           double relaxationIndex, double focusIndex, double signalStability, int sampleCount) {
            this.timestamp = timestamp;
            this.averageAttention = averageAttention;
            this.averageMeditation = averageMeditation;
            this.relaxationIndex = relaxationIndex;
            this.focusIndex = focusIndex;
            this.signalStability = signalStability;
            this.sampleCount = sampleCount;
        }
    }
    
    /**
     * 数据质量信息类
     */
    public static class DataQualityInfo {
        public final double signalQuality;
        public final double dataRate;
        public final boolean isStable;
        public final String status;
        
        public DataQualityInfo(double signalQuality, double dataRate, boolean isStable, String status) {
            this.signalQuality = signalQuality;
            this.dataRate = dataRate;
            this.isStable = isStable;
            this.status = status;
        }
    }
    
    /**
     * 构造函数
     */
    public DataProcessor() {
        this.dataBuffer = new ConcurrentLinkedQueue<>();
        this.rawDataBuffer = new ConcurrentLinkedQueue<>();
        this.executorService = Executors.newFixedThreadPool(2);
        this.dataSubject = PublishSubject.create();
        this.processedDataSubject = PublishSubject.create();
        this.qualitySubject = PublishSubject.create();
    }
    
    /**
     * 设置数据监听器
     */
    public void setDataProcessorListener(DataProcessorListener listener) {
        this.listener = listener;
    }
    
    /**
     * 开始数据处理
     */
    public void startProcessing(String sessionId, String deviceName, String deviceAddress) {
        this.currentSessionId = sessionId;
        this.deviceName = deviceName;
        this.deviceAddress = deviceAddress;
        this.isProcessing = true;
        this.totalSamples = 0;
        this.validSamples = 0;
        
        // 清空缓冲区
        dataBuffer.clear();
        rawDataBuffer.clear();
        
        // 启动数据处理任务
        startDataProcessingTask();
        startQualityMonitoringTask();
    }
    
    /**
     * 停止数据处理
     */
    public void stopProcessing() {
        this.isProcessing = false;
    }
    
    /**
     * 获取数据流Observable
     */
    public Observable<BrainWaveData> getDataStream() {
        return dataSubject.asObservable();
    }
    
    /**
     * 获取处理后数据流Observable
     */
    public Observable<ProcessedData> getProcessedDataStream() {
        return processedDataSubject.asObservable();
    }
    
    /**
     * 获取数据质量流Observable
     */
    public Observable<DataQualityInfo> getQualityStream() {
        return qualitySubject.asObservable();
    }
    
    // NeuroSkyDataListener 接口实现
    
    @Override
    public void onPoorSignalReceived(int poorSignal) {
        if (!isProcessing) return;
        
        // 更新当前数据的信号质量
        updateCurrentDataSignalQuality(poorSignal);
    }
    
    @Override
    public void onAttentionReceived(int attention) {
        if (!isProcessing) return;
        
        // 更新当前数据的注意力值
        updateCurrentDataAttention(attention);
    }
    
    @Override
    public void onMeditationReceived(int meditation) {
        if (!isProcessing) return;
        
        // 更新当前数据的冥想值
        updateCurrentDataMeditation(meditation);
    }
    
    @Override
    public void onBlinkReceived(int blinkStrength) {
        if (!isProcessing) return;
        
        // 更新当前数据的眨眼强度
        updateCurrentDataBlink(blinkStrength);
    }
    
    @Override
    public void onRawDataReceived(int rawValue) {
        if (!isProcessing) return;
        
        // 添加原始数据到缓冲区
        addRawData(rawValue);
    }
    
    @Override
    public void onEegPowerReceived(NeuroSkyProtocol.EegPowerData eegPower) {
        if (!isProcessing) return;
        
        // 创建完整的脑波数据
        BrainWaveData data = createBrainWaveData(eegPower);
        processData(data);
    }
    
    @Override
    public void onParseError(String error) {
        if (listener != null) {
            listener.onProcessingError("数据解析错误: " + error);
        }
    }
    
    /**
     * 创建脑波数据对象
     */
    private BrainWaveData createBrainWaveData(NeuroSkyProtocol.EegPowerData eegPower) {
        BrainWaveData data = new BrainWaveData();
        data.setTimestamp(System.currentTimeMillis());
        data.setSessionId(currentSessionId);
        data.setDeviceName(deviceName);
        data.setDeviceAddress(deviceAddress);
        
        // 设置EEG功率数据
        data.setDelta(eegPower.delta);
        data.setTheta(eegPower.theta);
        data.setLowAlpha(eegPower.lowAlpha);
        data.setHighAlpha(eegPower.highAlpha);
        data.setLowBeta(eegPower.lowBeta);
        data.setHighBeta(eegPower.highBeta);
        data.setLowGamma(eegPower.lowGamma);
        data.setHighGamma(eegPower.highGamma);
        
        return data;
    }
    
    /**
     * 处理脑波数据
     */
    private void processData(BrainWaveData data) {
        executorService.execute(() -> {
            try {
                // 数据验证
                if (validateData(data)) {
                    // 添加到缓冲区
                    addToBuffer(data);
                    
                    // 发送数据流
                    dataSubject.onNext(data);
                    
                    // 通知监听器
                    if (listener != null) {
                        listener.onDataProcessed(data);
                    }
                    
                    validSamples++;
                }
                
                totalSamples++;
                lastDataTime = System.currentTimeMillis();
                
            } catch (Exception e) {
                if (listener != null) {
                    listener.onProcessingError("数据处理错误: " + e.getMessage());
                }
            }
        });
    }
    
    /**
     * 验证数据有效性
     */
    private boolean validateData(BrainWaveData data) {
        // 检查信号质量
        if (data.getPoorSignal() > POOR_SIGNAL_THRESHOLD) {
            return false;
        }
        
        // 检查数据范围
        if (data.getAttention() < 0 || data.getAttention() > 100) {
            return false;
        }
        
        if (data.getMeditation() < 0 || data.getMeditation() > 100) {
            return false;
        }
        
        // 检查时间戳
        if (data.getTimestamp() <= 0) {
            return false;
        }
        
        data.setValid(true);
        return true;
    }
    
    /**
     * 添加数据到缓冲区
     */
    private void addToBuffer(BrainWaveData data) {
        dataBuffer.offer(data);
        
        // 保持缓冲区大小
        while (dataBuffer.size() > BUFFER_SIZE) {
            dataBuffer.poll();
        }
    }
    
    /**
     * 添加原始数据到缓冲区
     */
    private void addRawData(int rawValue) {
        rawDataBuffer.offer(rawValue);
        
        // 保持缓冲区大小
        while (rawDataBuffer.size() > RAW_BUFFER_SIZE) {
            rawDataBuffer.poll();
        }
    }
    
    /**
     * 启动数据处理任务
     */
    private void startDataProcessingTask() {
        executorService.execute(() -> {
            while (isProcessing) {
                try {
                    if (dataBuffer.size() >= MIN_VALID_SAMPLES) {
                        ProcessedData processedData = analyzeBufferedData();
                        if (processedData != null) {
                            processedDataSubject.onNext(processedData);
                            
                            if (listener != null) {
                                listener.onProcessedDataReady(processedData);
                            }
                        }
                    }
                    
                    Thread.sleep(1000); // 每秒处理一次
                    
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                } catch (Exception e) {
                    if (listener != null) {
                        listener.onProcessingError("数据分析错误: " + e.getMessage());
                    }
                }
            }
        });
    }
    
    /**
     * 启动质量监控任务
     */
    private void startQualityMonitoringTask() {
        executorService.execute(() -> {
            while (isProcessing) {
                try {
                    DataQualityInfo qualityInfo = calculateDataQuality();
                    qualitySubject.onNext(qualityInfo);
                    
                    if (listener != null) {
                        listener.onDataQualityChanged(qualityInfo);
                    }
                    
                    Thread.sleep(2000); // 每2秒检查一次质量
                    
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                } catch (Exception e) {
                    if (listener != null) {
                        listener.onProcessingError("质量监控错误: " + e.getMessage());
                    }
                }
            }
        });
    }
    
    /**
     * 分析缓冲区数据
     */
    private ProcessedData analyzeBufferedData() {
        List<BrainWaveData> dataList = new ArrayList<>(dataBuffer);
        if (dataList.isEmpty()) return null;
        
        // 计算平均值
        double avgAttention = dataList.stream().mapToInt(BrainWaveData::getAttention).average().orElse(0);
        double avgMeditation = dataList.stream().mapToInt(BrainWaveData::getMeditation).average().orElse(0);
        
        // 计算放松和专注指数
        double avgRelaxation = dataList.stream().mapToDouble(BrainWaveData::getRelaxationIndex).average().orElse(0);
        double avgFocus = dataList.stream().mapToDouble(BrainWaveData::getFocusIndex).average().orElse(0);
        
        // 计算信号稳定性
        double signalStability = calculateSignalStability(dataList);
        
        return new ProcessedData(
            System.currentTimeMillis(),
            avgAttention,
            avgMeditation,
            avgRelaxation,
            avgFocus,
            signalStability,
            dataList.size()
        );
    }
    
    /**
     * 计算信号稳定性
     */
    private double calculateSignalStability(List<BrainWaveData> dataList) {
        if (dataList.size() < 2) return 0.0;
        
        // 计算信号质量的标准差
        double avgPoorSignal = dataList.stream().mapToInt(BrainWaveData::getPoorSignal).average().orElse(0);
        double variance = dataList.stream()
            .mapToDouble(data -> Math.pow(data.getPoorSignal() - avgPoorSignal, 2))
            .average().orElse(0);
        
        double stdDev = Math.sqrt(variance);
        
        // 稳定性 = 100 - 标准差百分比
        return Math.max(0, 100 - (stdDev / avgPoorSignal * 100));
    }
    
    /**
     * 计算数据质量
     */
    private DataQualityInfo calculateDataQuality() {
        // 信号质量 (基于最近的数据)
        double signalQuality = 0.0;
        if (!dataBuffer.isEmpty()) {
            List<BrainWaveData> recentData = new ArrayList<>(dataBuffer);
            Collections.reverse(recentData);
            
            // 取最近10个样本的平均信号质量
            int sampleCount = Math.min(10, recentData.size());
            double avgPoorSignal = recentData.stream()
                .limit(sampleCount)
                .mapToInt(BrainWaveData::getPoorSignal)
                .average().orElse(200);
            
            signalQuality = Math.max(0, 100 - (avgPoorSignal / 2.0));
        }
        
        // 数据速率 (样本/秒)
        double dataRate = 0.0;
        long currentTime = System.currentTimeMillis();
        if (lastDataTime > 0 && totalSamples > 0) {
            double timeSpan = (currentTime - (currentTime - 10000)) / 1000.0; // 最近10秒
            dataRate = validSamples / Math.max(1, timeSpan);
        }
        
        // 稳定性判断
        boolean isStable = signalQuality > 70 && dataRate > 0.5;
        
        // 状态描述
        String status;
        if (signalQuality > 80) {
            status = "优秀";
        } else if (signalQuality > 60) {
            status = "良好";
        } else if (signalQuality > 40) {
            status = "一般";
        } else {
            status = "较差";
        }
        
        return new DataQualityInfo(signalQuality, dataRate, isStable, status);
    }
    
    /**
     * 更新当前数据的信号质量
     */
    private void updateCurrentDataSignalQuality(int poorSignal) {
        // 这里可以实现更复杂的逻辑来更新当前正在构建的数据对象
    }
    
    /**
     * 更新当前数据的注意力值
     */
    private void updateCurrentDataAttention(int attention) {
        // 这里可以实现更复杂的逻辑来更新当前正在构建的数据对象
    }
    
    /**
     * 更新当前数据的冥想值
     */
    private void updateCurrentDataMeditation(int meditation) {
        // 这里可以实现更复杂的逻辑来更新当前正在构建的数据对象
    }
    
    /**
     * 更新当前数据的眨眼强度
     */
    private void updateCurrentDataBlink(int blinkStrength) {
        // 这里可以实现更复杂的逻辑来更新当前正在构建的数据对象
    }
    
    /**
     * 获取缓冲区统计信息
     */
    public String getBufferStats() {
        return String.format("缓冲区: %d/%d, 原始数据: %d/%d, 总样本: %d, 有效样本: %d",
            dataBuffer.size(), BUFFER_SIZE,
            rawDataBuffer.size(), RAW_BUFFER_SIZE,
            totalSamples, validSamples);
    }
    
    /**
     * 清理资源
     */
    public void cleanup() {
        isProcessing = false;
        dataBuffer.clear();
        rawDataBuffer.clear();
        
        if (executorService != null && !executorService.isShutdown()) {
            executorService.shutdown();
        }
        
        dataSubject.onComplete();
        processedDataSubject.onComplete();
        qualitySubject.onComplete();
    }
}